const EventEmitter = require("events");
const fs = require("fs");
class ReadStream extends EventEmitter {
  constructor(path, options = {}) {
    super();
    this.path = path;
    this.flags = options.flags || "r";
    this.autoClose = options.autoClose || true;
    this.emitClose = options.emitClose || true;
    this.start = options.start || 0;
    this.offset = this.start;
    this.highWaterMark = options.highWaterMark || 64 * 1024;
    this.flowing = false; // 默认为非流动模式， 用户没有读取数据
    this.open(); // 这个打开操作是异步的
    this.on("newListener", function (eventName) {
      if (eventName === "data") {
        // 用户绑定了data事件
        this.flowing = true;
        this.read(); // 需要用到打开的fd
      }
    });
  }
  destroy(err) {
    if (typeof this.fd == "number") {
      if (this.autoClose) {
        fs.close(this.fd, () => {
          if (this.emitClose) {
            this.emit("close");
          }
        });
      }
    }
    if (err) {
      this.emit("error", err);
    }
  }
  pause() {
    this.flowing = false;
  }
  resume() {
    if (!this.flowing) {
      this.flowing = true;
      this.read();
    }
  }
  read() {
    if (typeof this.fd !== "number") {
      return this.once("open", this.read);
    }
    this.buffer = Buffer.alloc(this.highWaterMark);
    fs.read(
      this.fd,
      this.buffer,
      0,
      this.highWaterMark,
      this.offset,
      (err, bytesRead) => {
        if (err) {
          return this.destroy(err);
        }
        if (bytesRead) {
          this.offset += bytesRead;
          this.emit("data", this.buffer.slice(0, bytesRead));
          if (this.flowing) {
            this.read(); // 递归读取
          }
        } else {
          this.emit("end");
          this.destroy();
        }
      }
    );
  }
  open(ws) {
    fs.open(this.path, this.flags, (err, fd) => {
      if (err) {
        return this.destroy(err);
      }
      this.fd = fd; // 打开成功后就有了fd
      this.emit("open", fd);
    });
  }
  pipe(ws) {
    this.on("data", (chunk) => {
      let flag = ws.write(chunk); //  写入的返回值来控制读取的速率， 实现了 读写间的分离
      if (!flag) {
        this.pause();
      }
    });
    ws.on("drain", () => {
      this.resume();
    });
    this.on("end", function () {
      ws.end();
    });
  }
}

module.exports = ReadStream;
